package com.diaohaiwen.framework.mq.rocket.config;

import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSONObject;
import com.diaohaiwen.framework.mq.rocket.RocketService;
import com.diaohaiwen.framework.mq.rocket.attrbute.*;
import com.diaohaiwen.framework.mq.rocket.constant.RocketHeaderConst;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.util.List;
import java.util.Map;

import static com.diaohaiwen.framework.mq.rocket.constant.RocketTagConstant.defaultTag;

/**
 * @author 刁海文
 * @since 2024/1/15 23:13
 */
@Slf4j
@Component
public class GlobalMessageListener implements MessageListenerConcurrently {


    @Resource
    RocketService rocketService;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        try {
            for (MessageExt ext : list) {
                Method method = findMethod(ext.getTopic(), ext.getTags());

                Object invoke = invoke(method, ext);

                sendTo(method, invoke);
            }

            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

        } catch (Exception e) {
            e.printStackTrace();
            for (MessageExt message : list) {
                log.error("Topic:[{}], Tag:[{}], Body:{}", message.getTopic(), message.getTags(), new String(message.getBody()));

            }
        }
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }


    private Method findMethod(String topic, String tag) {

        Map<String, Method> map = METHOD_MAP.get(topic);
        if (map == null) {
            return null;
        }
        if (map.containsKey(tag)) {
            return map.get(tag);
        }
        return map.get(defaultTag);
    }


    private Object invoke(Method method, MessageExt message) throws InvocationTargetException, IllegalAccessException {
        if (method == null) {
            log.error("Topic:[{}], Tag:[{}] is not found.", message.getTopic(), message.getTags());
            return null;
        }

        Parameter[] parameters = method.getParameters();

        Object[] objects = new Object[parameters.length];
        for (int i = 0; i < parameters.length; i++) {
            Parameter parameter = parameters[i];
            if (parameter.isAnnotationPresent(RocketBody.class)) {
                objects[i] = JSONObject.parseObject(message.getBody(), parameter.getParameterizedType());
            } else if (parameter.isAnnotationPresent(RocketHeader.class)) {
                String key = parameter.getAnnotation(RocketHeader.class).value();
                objects[i] = getHeader(key, message);
            } else if (parameter.isAnnotationPresent(RocketHeaders.class)) {
                RocketHeaders anno = parameter.getAnnotation(RocketHeaders.class);
                JSONObject headers = getHeaders(anno, message);
                objects[i] = headers.toJavaObject(parameter.getParameterizedType());
            }
        }
        Object bean = SpringUtil.getBean(method.getDeclaringClass());
        return method.invoke(bean, objects);
    }


    private Object getHeader(String key, MessageExt ext) {

        return switch (key) {
            case RocketHeaderConst.keys -> ext.getKeys();
            case RocketHeaderConst.msgId -> ext.getMsgId();
            case RocketHeaderConst.preparedTransactionOffset -> ext.getPreparedTransactionOffset();
            case RocketHeaderConst.bodyCRC -> ext.getBodyCRC();
            case RocketHeaderConst.bornHost -> ext.getBornHost();
            case RocketHeaderConst.bornTimestamp -> ext.getBornTimestamp();
            case RocketHeaderConst.commitLogOffset -> ext.getCommitLogOffset();
            case RocketHeaderConst.brokerName -> ext.getBrokerName();
            case RocketHeaderConst.queueId -> ext.getQueueId();
            case RocketHeaderConst.queueOffset -> ext.getQueueOffset();
            case RocketHeaderConst.storeHost -> ext.getStoreHost();
            case RocketHeaderConst.storeSize -> ext.getStoreSize();
            case RocketHeaderConst.storeTimestamp -> ext.getStoreTimestamp();
            case RocketHeaderConst.sysFlag -> ext.getSysFlag();
            case RocketHeaderConst.reconsumeTimes -> ext.getReconsumeTimes();
            default -> ext.getProperty(key);
        };

    }


    private JSONObject getHeaders(RocketHeaders anno, MessageExt ext) {
        JSONObject params = new JSONObject();

        params.putAll(ext.getProperties());
        if (anno.value()) {
            String[] attributes = anno.attrbutes().length == 0 ? RocketHeaderConst.systemProperties() : anno.attrbutes();
            for (String property : attributes) {
                params.put(property, getHeader(property, ext));
            }
        }
        return params;
    }

    private void sendTo(Method method, Object val) {
        if (val == null || !method.isAnnotationPresent(RocketSendTo.class)) {
            return;
        }

        RocketSendTo send = method.getAnnotation(RocketSendTo.class);

        rocketService.send(send.topics(), send.tags(), val);
    }
}
